import org.apache.spark.sql.Row
import scala.collection.mutable.{Map => MutableMap}
/**
 * 定义Transformer Trait，所有Transformer需实现transform方法
 */
trait Transformer {
  def transform(dt: String, hour: String, ifTest: Boolean = false): Unit
}


/**
 * 共享类型定义
 * 用于在多个类之间共享的数据结构
 */

/**
 * 操作符特征
 * @param id 特征ID
 * @param value 特征值
 */
case class OperatorFeature(id: Long, value: String)

/**
 * 操作符特征集合
 * @param op_name 操作符名称
 * @param features 特征列表
 * @param updated_time 更新时间
 */
case class OperatorFeatures(op_name: String, features: Seq[OperatorFeature], updated_time: Long)

/**
 * trace表的记录
 * @param trace_id 追踪ID
 * @param account_id 账户ID
 * @param day_part 日期分区
 * @param operator_features 操作符特征列表
 */
case class TraceRecord(
  trace_id: Long,
  account_id: Long,
  day_part: Long,
  operator_features: Seq[OperatorFeatures],
  calc_time: Long
)

/**
 * 人时表的记录
 * @param account_id 账户ID
 * @param fetch_time 数据计算的时刻
 * @param day_part 日期分区
 * @param operator_features 操作符特征列表
 */
case class AccountTimeRecord(
  account_id: Long,
  fetch_time: Long,
  op_name: String,
  op_features: Map[Long, String],
  updated_time: Long,
  day_part: Long,
)


/**
 * 人时表历史记录
 * @param account_id 账户ID
 * @param history_start 历史开始时间，即min(op_history) = op_history.head
 * @param history_end 历史结束时间, 如果freeze is false, 则history_end = +inf,否则history_end = max(op_history) = op_history.last
 * @param op_history 操作符历史记录，排序好的时间戳
 */
// case class AccountTimeIndexRecord(
//   account_id: Long,
//   var history_start: Long = 0,
//   var history_end: Long = 0,
//   op_name: String,
//   var op_history: Seq[Long] = Seq.empty[Long],
//   day_part: Long
// ) 

/**
  * 上面那个版本需要通过merge into保证数据的完备性
  * spark开销会越来越大。短期先支持这个版本，每次只是简单insert into。
  * 查询的时候通过max_by查询，获得时间戳和day_part
  * 
  * @param account_id
  * @param op_name
  * @param fetch_time
  * @param day_part
  */
case class AccountTimeIndexRecord(
  account_id: Long,
  op_name: String,
  fetch_time: Long,
  day_part: Long
) 

/**
 * 输出的特征信息
 * @param value 特征值
 * @param updated_time 更新时间
 */
case class FeatureInfo(value: String, updated_time: Long)

/**
 * 类型转换工具对象
 */
object TypeConverters {
  
  /**
   * 从Row转换为OperatorFeature
   */
  def rowToOperatorFeature(row: Row): OperatorFeature = {
    OperatorFeature(
      id = row.getAs[Long](0),
      value = row.getAs[String](1)
    )
  }
  
  /**
   * 从Row转换为OperatorFeatures
   */
  def rowToOperatorFeatures(row: Row): Option[OperatorFeatures] = {
    val opName = row.getAs[String](0)
    val featuresRow = Option(row.getAs[Seq[Row]](1))
    val updateTime = row.getAs[Long](2)
    featuresRow match {
      case None => None
      case Some(frow) => {
        val features = frow.map(rowToOperatorFeature)
        Some(OperatorFeatures(
          op_name = opName,
          features = features,
          updated_time = updateTime
        ))
      }
    }
  }
  
  /**
   * 从Row转换为TraceRecord
   */
  def rowToTraceRecord(row: Row): TraceRecord = {
    val traceId = row.getAs[Long]("trace_id")
    val accountId = row.getAs[Long]("account_id")
    val dayPart = row.getAs[Long]("day_part")
    val operatorFeaturesRow = row.getAs[Seq[Row]]("operator_features")
    val calcTime = row.getAs[Long]("calc_time")
    
    val operatorFeatures = operatorFeaturesRow.flatMap(features => 
      rowToOperatorFeatures(features) match {
        case Some(of) => Some(of)
        case None => {
          print(s"WARN: operatorFeatures is null, traceId: $traceId, accountId: $accountId, dayPart: $dayPart")
          None
        }
      }
    )
    
    TraceRecord(
      trace_id = traceId,
      account_id = accountId,
      day_part = dayPart,
      operator_features = operatorFeatures,
      calc_time = calcTime
    )
  }
  /**
   * 从Row转换为AccountRecord
   */
  def rowToAccountRecord(row: Row): Seq[AccountTimeRecord] = {
    val accountId = row.getAs[Long]("account_id")
    val fetchTime = row.getAs[Long]("fetch_time")
    val dayPart = row.getAs[Long]("day_part")
    val operatorFeaturesRow = row.getAs[Seq[Row]]("operator_features")
    val operatorFeatures = operatorFeaturesRow.flatMap(features => 
      rowToOperatorFeatures(features) match {
        case Some(of) => Some(AccountTimeRecord(
          account_id = accountId,
          fetch_time = fetchTime,
          day_part = dayPart,
          op_name = of.op_name.toLowerCase(),
          updated_time = of.updated_time,
          op_features = of.features.map { feature =>
            (feature.id, feature.value)
          }.toMap
        ))
        case None => {
          println(s"WARN: operatorFeatures is null, accountId: $accountId, fetchTime: $fetchTime, dayPart: $dayPart")
          None
        }
      }
    )
    operatorFeatures
  }

  /**
   * 从TraceRecord转换为Row
   */
  def traceRecordToRow(record: TraceRecord): Row = {
    Row(
      record.trace_id,
      record.account_id,
      record.day_part,
      record.operator_features
    )
  }
  
}